package com.example.demo.consu;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;


import java.time.Duration;
import java.util.*;

/**
 *  自定义提交offset ，当然kafka也提供了同步和异步的提交方式,但存在数据漏消费或者重复消费
 *
 */
public class MyConsumer2 {


    private static final Map<TopicPartition,Long> currentOffset = new HashMap<>();


    public static void main(String[] args) {

        Properties properties = new Properties();

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.1");

        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserialization");

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserialization");

        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"cwy");

        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

        consumer.subscribe(Arrays.asList("first", "111"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 此方法Rebalance 之前调用
                commitOffset(currentOffset);
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // 此方法Rebalance之后调用
                for (TopicPartition partition:
                     partitions) {
                    consumer.seek(partition,getOffset(partition));
                }
            }
        });
        while (true){
            ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String,String> record:
                 records) {
                commitOffset(currentOffset);
            }
        }
    }

    private static void commitOffset(Map<TopicPartition,Long> map){
        // 自己实现逻辑
    }

    private static long getOffset(TopicPartition partition){
        return 0; // 自己实现逻辑
    }
}
